package com.atguigu.flink.wordcount;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Created by Smexy on 2022/11/19
 *
 *      OperatorChain:  算子链。  把多个算子链成一个 算子
 *
 *      有条件:
 *                  ①算子的上下游分发规则必须是 forward(1:1)
 *                          不能是 1：N
 *                          也不能是 N：1
 *                  ②上下游的并行度一致
 *                  ③只有处于同一个 SlotSharingGroup的才能合并
 *                          默认所有的算子的SlotSharingGroup都是一样的！
 *
 *                          不同的算子需要的资源可能是不同的，有的算子比较消费内存，有的不消耗内存。
 *                          在合并时，不建议把性质相同的算子合并！
 *                              A：  消耗内存
 *                              B： 消费内存
 *
 *                              合并 (A,B)是不合适的！
 *
 *                             SlotSharingGroup是一种标记，只有SlotSharingGroup相同的，才能进行 OperatorChain的合并！
 *                             如果两个算子的SlotSharingGroup不同，无法合并！即使满足了 forward条件！
 *
 *                             当修改了SlotSharingGroup，后续的算子都会使用此标记！
 *
 *
 *
 *      默认开启OperatorChain，有什么好处?
 *              1.节省资源，省slot，节省网络IO
 *              2.提高效率，多个算子在一个Task中串行运行
 *
 *       某种情况下，也可以禁止算子链合并！
 *                  1.全局禁止  /env.disableOperatorChaining();
 *                  2.使当前算子和前后的算子断开   算子.disableChaining()
 *                  3.使当前算子和前面的算子断开   算子.startNewChain()
 *
 *

 */
public class Demo11_OperatorChain
{
    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port",3333);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);

        //全局禁止OperatorChain
        //env.disableOperatorChaining();

        

        DataStreamSource<String> ds = env.socketTextStream("hadoop103", 8888).setParallelism(1);

        ds
            .map(s -> s).setParallelism(1)
            .map(s -> s).setParallelism(1).name("map 断开").slotSharingGroup("a")
            .map(s -> s).setParallelism(1).slotSharingGroup("a")
            .map(s -> s).setParallelism(1).slotSharingGroup("b")//.startNewChain()//.disableChaining()
            //.global()  //N:1
            //.shuffle()  //1:N
            .print().setParallelism(1);


        env.execute();

    }
}
